package com.yide.myworld.service;

import com.alibaba.fastjson2.JSON;
import com.yide.myworld.domain.MQMessageDataDto;
import com.yide.myworld.domain.MqFlow;
import com.yide.myworld.mapper.MqFlowMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;

import javax.annotation.Resource;
import java.sql.Timestamp;
import java.util.concurrent.TimeUnit;

@Slf4j
public abstract class AbstractMQConsumer implements RocketMQListener<MQMessageDataDto> {

    @Resource
    private MqFlowMapper mapper;
    @Resource
    private RedissonClient redissonClient;

    @Override
    public void onMessage(MQMessageDataDto message) {
        long startTime = System.currentTimeMillis();
        Class<? extends AbstractMQConsumer> aClass = this.getClass();
        RocketMQMessageListener annotation = aClass.getAnnotation(RocketMQMessageListener.class);
        String topic = annotation.topic();
        String messageId = message.getMessageId();
        RLock lock = redissonClient.getLock(messageId);
        try {
            if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {

                log.info("处理业务，message:[{}],topic:[{}]", JSON.toJSONString(message), topic);
                MqFlow flow = mapper.selectById(messageId);
                if (ObjectUtils.isNotEmpty(flow)) {
                    if ("success".equals(flow.getResult())) {
                        log.error("mq幂等校验失败！messageId:[{}]", messageId);
                        return;
                    }
                }
                log.info("幂等校验成功，执行业务逻辑！");
                MqFlow mqFlow = new MqFlow();
                mqFlow.setMessageId(messageId);
                mqFlow.setTopic(topic);
                mqFlow.setMessage(message.getData());
                mqFlow.setBusinessTime(new Timestamp(startTime));
                try {
                    doService(message.getData());
                } catch (Exception e) {
                    log.error("mq消费失败，messageId:[{}],topic:[{}],message:[{}]", messageId, topic, JSON.toJSONString(message.getData()));
                    mqFlow.setResult(e.getMessage());
                    if (ObjectUtils.isNotEmpty(flow)) {
                        mapper.updateByPrimaryKeySelective(mqFlow);
                    } else {
                        mapper.insert(mqFlow);
                    }
                    throw e;
                }
                mqFlow.setResult("success");
                if (ObjectUtils.isNotEmpty(flow)) {
                    mapper.updateByPrimaryKeySelective(mqFlow);
                } else {
                    mapper.insert(mqFlow);
                }
                log.info("mq消费成功，messageId:[{}],topic:[{}],message:[{}]", messageId, topic, JSON.toJSONString(message.getData()));
            }
        } catch (InterruptedException e) {
            log.error("消費失败，【{}】", messageId);
            throw new RuntimeException(e);
        } finally {
            if (lock.isHeldByCurrentThread()) {
                lock.unlock();
                log.info("释放锁成功，messageI：【{}】", messageId);
            }
        }
    }

    public abstract void doService(String message) throws InterruptedException;
}
